package com.basker.pisces.redis.counter;

import org.springframework.dao.DataRetrievalFailureException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;

import com.basker.pisces.concurrent.counter.ICounter;

class RedisCounter implements ICounter {

    private volatile String key;

    private final ValueOperations<String, Integer> operations;
    private final RedisOperations<String, Integer> generalOps;

    public RedisCounter(String key, RedisOperations<String, Integer> template) {
        this(key, template, null);
    }

    public RedisCounter(String key, RedisOperations<String, Integer> template, int initialValue) {
        this(key, template, Integer.valueOf(initialValue));
    }

    private RedisCounter(String key, RedisOperations<String, Integer> template, @Nullable Integer initialValue) {

        Assert.hasText(key, "parameter 'key' is required");
        Assert.notNull(template, "parameter 'template' is required");
        Assert.notNull(template.getKeySerializer(), "parameter 'template.getKeySerializer()' is required");
        Assert.notNull(template.getValueSerializer(), "parameter 'template.getValueSerializer()' is required");

        this.key = key;
        this.generalOps = template;
        this.operations = generalOps.opsForValue();

        if (initialValue == null) {
            initial(0);
        } else {
            initial(initialValue);
        }
    }

    @Override
    public int decrementAndGet() {
        return operations.increment(key, -1).intValue();
    }

    @Override
    public int get() {
        Integer value = operations.get(key);
        if (value != null) {
            return value;
        }

        throw new DataRetrievalFailureException(String.format("The key '%s' seems to no longer exist.", key));
    }

    @Override
    public int getAndDecrement() {
        return decrementAndGet() + 1;
    }

    @Override
    public int getAndIncrement() {
        return incrementAndGet() - 1;
    }

    @Override
    public String getKey() {
        return key;
    }

    @Override
    public int incrementAndGet() {
        return operations.increment(key, 1).intValue();
    }

    @Override
    public void set(int value) {
        operations.set(key, value);
    }

    private void initial(int value) {
        operations.setIfAbsent(key, value);
    }

}
